package com.yjjxt

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Hello07Join {
  def main(args: Array[String]): Unit = {
    //1.配置并创建对象
    val sparkContext = new SparkContext((new SparkConf().setMaster("local").setAppName("Join" + System.currentTimeMillis())))
    //读取数据
    //    val array = Array[String]("Hello1 user1", "Hello2 user2", "Hello user", "Hello user", "user1 apple")
    //    val lines: RDD[String] = sparkContext.parallelize(array, 8)
    //    println(lines.getNumPartitions)
    //读取数据
    //    val linesPart: RDD[String] = sparkContext.textFile("src/main/resources/part.txt", 4)
    //    println(linesPart.getNumPartitions)

    //开始关联数据 `(K,V)join(K,W)返回(K,(V,W))`
    val array1 = Array[String]("Hello1 user1", "Hello2 user1", "Hello user11", "Hello user12", "user1 apple1")
    val array2 = Array[String]("Hello1 user2", "Hello2 user2", "Hello user21", "Hello user22", "user2 apple2")
    val lines1 = sparkContext.parallelize(array1, 3)
    val lines2 = sparkContext.parallelize(array2, 4)
    val words1 = lines1.map(ele => (ele.split(" ")(0), ele.split(" ")(1)))
    val words2 = lines2.map(ele => (ele.split(" ")(0), ele.split(" ")(1)))

    //Join(按照Key等值关联)
    words1.join(words2).foreach(ele => println("join=>" + ele))
    //leftOuterJoin (按照Key等值关联,然后显示左面不满足条件的)
    words1.leftOuterJoin(words2).foreach(ele => println("leftOuterJoin=>" + ele))
    //rightOuterJoin (按照Key等值关联,然后显示右面不满足条件的)
    words1.rightOuterJoin(words2).foreach(ele => println("rightOuterJoin=>" + ele))
    //fullOuterJoin (按照Key等值关联,然后显示左右所有不满足条件的)
    words1.fullOuterJoin(words2).foreach(ele => println("fullOuterJoin=>" + ele))

    //查看分区数
    println("合并后的分区数join：" + words1.join(words2, 5).getNumPartitions)
    println("合并后的分区数leftOuterJoin：" + words1.leftOuterJoin(words2, 7).getNumPartitions)

  }
}
